package com.katze.boot.common.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 所有线程池初始化
 */
@EnableAsync
@Configuration
public class ExecutorConfigurer{
    private static final Logger log = LoggerFactory.getLogger(ExecutorConfigurer.class);

    private ThreadPoolTaskExecutor taskExecutor;
    private ThreadPoolExecutor sseEmitterPoolExecutor;

    /**
     * 为@Async标签提供使用,详情：AsyncExecutionAspectSupport.getDefaultExecutor()
     * @return 异步线程池
     */
    @Bean(value = "taskExecutor")
    public TaskExecutor taskExecutor() {
        taskExecutor = new ThreadPoolTaskExecutor();

        taskExecutor.setCorePoolSize(5);
        taskExecutor.setMaxPoolSize(100);
        taskExecutor.setQueueCapacity(1000);
        taskExecutor.setKeepAliveSeconds(1);
        taskExecutor.setThreadNamePrefix("AsyncTaskExecutor-");

        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        taskExecutor.initialize();
        return taskExecutor;
    }

    /**
     * 处理文件下载进度
     */
    @Bean(value = "sftpSseEmitterPoolExecutor")
    public ThreadPoolExecutor sseEmitterPoolExecutor() {
        sseEmitterPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
        sseEmitterPoolExecutor.setThreadFactory(new CustomizableThreadFactory("SFTP-SseEmitter-"));
        return sseEmitterPoolExecutor;
    }


    @EventListener
    public void onContextClosedEvent(ContextClosedEvent event) throws InterruptedException {
        log.info("The application is preparing to close all thread pools...");


        log.info("准备关闭线程池SFTP-SseEmitter-x,当前任务量: {}", sseEmitterPoolExecutor.getQueue().size());
        sseEmitterPoolExecutor.shutdownNow();
        log.info("SFTP-SseEmitter-x已关闭.");

        log.info("准备关闭线程池AsyncTaskExecutor-x当前任务量: {}", taskExecutor.getQueueSize());
        while (taskExecutor.getQueueSize() != 0) {
            TimeUnit.SECONDS.sleep(1);
            log.info("剩余任务量：{}", taskExecutor.getQueueSize());
        }
        log.info("AsyncTaskExecutor-x已关闭.");
    }
}
